#include "ZDownloader.h"
#include "ZLog.h"
#include <curl/curl.h>
#include <unordered_map>
#include "ZFileUtils.h"
#include "SDL.h"

namespace ZuiLib {

	static const long LOW_SPEED_LIMIT = 1;
	static const long LOW_SPEED_TIME = 15;



	const std::string& DownloadTask::GetPath()const
	{
		return m_storagePath;
	}
	const std::string& DownloadTask::GetUrl()const
	{
		return m_requestURL;
	}


	class CurlDownloadTask:public DownloadTask
	{
	public:
		static std::set<std::string> s_storagePathSet;
		CurlDownloadTask(Downloader* owner)
			: m_fp(NULL),m_owner(owner)
		{
			_initInternal();
		}

		virtual ~CurlDownloadTask()
		{
			if (m_tempFileName.length() && s_storagePathSet.end() != s_storagePathSet.find(m_tempFileName))
			{
				s_storagePathSet.erase(m_tempFileName);
			}

			if (m_fp)
			{
				fclose(m_fp);
				m_fp = NULL;
			}
		}

		bool init(const std::string& filename, const std::string& tempSuffix);
		void initProc();
		void _initInternal();
		void setErrorProc(int code, int codeInternal, const char *desc);
		
		static size_t _outputHeaderCallbackProc(void *buffer, size_t size, size_t count, void *userdata);
		static size_t _outputDataCallbackProc(void *buffer, size_t size, size_t count, void *userdata);
		void _initCurlHandleProc(CURL *handle,bool forContent = false);
		bool _getHeaderInfoProc(CURL *handle);
		size_t writeDataProc(unsigned char *buffer, size_t size, size_t count);


		// header info
		bool    m_acceptRanges;
		bool    m_headerAchieved;
		int64_t m_totalBytesExpected;

		std::string  m_header;        // temp buffer for receive header string, only used in thread proc

		// progress
		int64_t m_bytesReceived;
		int64_t m_totalBytesReceived;

		// error
		int     m_errCodeInternal;
		std::string  m_errDescription;

		// for saving data
		std::string m_fileName;
		std::string m_tempFileName;
		std::vector<unsigned char> m_buf;
		FILE*  m_fp;

		ZMutex m_mutex;

		Downloader* m_owner;

	};

	typedef ZRefCountedPtr<CurlDownloadTask> CurlDownloadTaskPtr;
	std::set<std::string> CurlDownloadTask::s_storagePathSet;

	bool CurlDownloadTask::init(const  std::string& filename, const  std::string& tempSuffix)
	{
		if (0 == filename.length())
		{
			// data task
			m_buf.reserve(CURL_MAX_WRITE_SIZE);
			return true;
		}

		// file task
		m_fileName = filename;
		m_tempFileName = filename;
		m_tempFileName.append(tempSuffix);

		if (s_storagePathSet.end() != s_storagePathSet.find(m_tempFileName))
		{
			// there is another task uses this storage path
			m_errCode = DownloadTask::ERROR_FILE_OP_FAILED;
			m_errCodeInternal = 0;
			m_errDescription = "More than one download file task write to same file:";
			m_errDescription.append(m_tempFileName);
			return false;
		}
		s_storagePathSet.insert(m_tempFileName);

		// open temp file handle for write
		bool ret = false;
		do
		{
			std::string dir;
			size_t found = m_tempFileName.find_last_of("/\\");
			if (found == std::string::npos)
			{
				m_errCode = DownloadTask::ERROR_INVALID_PARAMS;
				m_errCodeInternal = 0;
				m_errDescription = "Can't find dirname in storagePath.";
				break;
			}

			// ensure directory is exist
			ZFileUtils* util = ZFileUtils::Instance();
			dir = m_tempFileName.substr(0, found+1);
			if (false == util->IsDirectoryExist(dir.c_str()))
			{
				if (false == util->CreateDirectory(dir.c_str()))
				{
					m_errCode = DownloadTask::ERROR_FILE_OP_FAILED;
					m_errCodeInternal = 0;
					m_errDescription = "Can't create dir:";
					m_errDescription.append(dir);
					break;
				}
			}

			// open file
#ifdef WIN32
			std::wstring file = Utf8ToUnicode(m_tempFileName);
			m_fp = _wfopen(file.c_str(), L"wb");
#else
			m_fp = fopen(m_tempFileName.c_str(), "wb");
#endif
			if (NULL == m_fp)
			{
				m_errCode = DownloadTask::ERROR_FILE_OP_FAILED;
				m_errCodeInternal = 0;
				m_errDescription = "Can't open file:";
				m_errDescription.append(m_tempFileName);
			}
			ret = true;
		} while (0);

		return ret;
	}

	void CurlDownloadTask::_initInternal()
	{
		m_acceptRanges = (false);
		m_headerAchieved = (false);
		m_bytesReceived = (0);
		m_totalBytesReceived = (0);
		m_totalBytesExpected = (0);
		m_errCode = (DownloadTask::ERROR_NO_ERROR);
		m_errCodeInternal = (CURLE_OK);
		m_header.resize(0);
		m_header.reserve(384);   // pre alloc header string buffer
	}


	void CurlDownloadTask::initProc()
	{
		ZAutoLock lock(&m_mutex);
		_initInternal();
	}

	void CurlDownloadTask::setErrorProc(int code, int codeInternal, const char *desc)
	{
		ZAutoLock lock(&m_mutex);
		m_errCode = code;
		m_errCodeInternal = codeInternal;
		m_errDescription = desc;
	}

	size_t CurlDownloadTask::writeDataProc(unsigned char *buffer, size_t size, size_t count)
	{
		ZAutoLock lock(&m_mutex);
		size_t ret = 0;
		if (m_fp)
		{
			ret = fwrite(buffer, size, count, m_fp);
		}
		else
		{
			ret = size * count;
			size_t cap = m_buf.capacity();
			size_t bufSize = m_buf.size();
			if (cap < bufSize + ret)
			{
				m_buf.reserve(bufSize * 2);
			}
			m_buf.insert(m_buf.end() , buffer, buffer + ret);
		}
		if (ret)
		{
			m_bytesReceived += ret;
			m_totalBytesReceived += ret;
		}
		return ret;
	}


	size_t CurlDownloadTask::_outputHeaderCallbackProc(void *buffer, size_t size, size_t count, void *userdata)
	{
		int strLen = int(size * count);
		CurlDownloadTask& coTask = *((CurlDownloadTask*)(userdata));
		coTask.m_header.append((const char *)buffer, strLen);
		return strLen;
	}

	size_t CurlDownloadTask::_outputDataCallbackProc(void *buffer, size_t size, size_t count, void *userdata)
	{
		//            DLLOG("    _outputDataCallbackProc: size(%ld), count(%ld)", size, count);
		CurlDownloadTask *coTask = (CurlDownloadTask*)userdata;

		// If your callback function returns CURL_WRITEFUNC_PAUSE it will cause this transfer to become paused.
		return coTask->writeDataProc((unsigned char *)buffer, size, count);
	}

	// this function designed call in work thread
	// the curl handle destroyed in _threadProc
	// handle inited for get header
	void CurlDownloadTask::_initCurlHandleProc(CURL *handle,bool forContent)
	{
		// set url
		curl_easy_setopt(handle, CURLOPT_URL,m_requestURL.c_str());

		// set write func
		if (forContent)
		{
			curl_easy_setopt(handle, CURLOPT_WRITEFUNCTION, _outputDataCallbackProc);
		}
		else
		{
			curl_easy_setopt(handle, CURLOPT_WRITEFUNCTION, _outputHeaderCallbackProc);
		}
		curl_easy_setopt(handle, CURLOPT_WRITEDATA, this);

		curl_easy_setopt(handle, CURLOPT_NOPROGRESS, true);
		//            curl_easy_setopt(handle, CURLOPT_XFERINFOFUNCTION, DownloaderCURL::Impl::_progressCallbackProc);
		//            curl_easy_setopt(handle, CURLOPT_XFERINFODATA, coTask);

		curl_easy_setopt(handle, CURLOPT_FAILONERROR, true);
		curl_easy_setopt(handle, CURLOPT_NOSIGNAL, 1L);

		if (forContent)
		{
			/** if server acceptRanges and local has part of file, we continue to download **/
			if (m_acceptRanges && m_totalBytesReceived > 0)
			{
				curl_easy_setopt(handle, CURLOPT_RESUME_FROM_LARGE,(curl_off_t)m_totalBytesReceived);
			}
		}
		else
		{
			// get header options
			curl_easy_setopt(handle, CURLOPT_HEADER, 1);
			curl_easy_setopt(handle, CURLOPT_NOBODY, 1);
		}

		//            if (!sProxy.empty())
		//            {
		//                curl_easy_setopt(curl, CURLOPT_PROXY, sProxy.c_str());
		//            }
		if (m_owner->getHints()->timeoutInSeconds)
		{
			curl_easy_setopt(handle, CURLOPT_CONNECTTIMEOUT, m_owner->getHints()->timeoutInSeconds);
		}


		curl_easy_setopt(handle, CURLOPT_LOW_SPEED_LIMIT, LOW_SPEED_LIMIT);
		curl_easy_setopt(handle, CURLOPT_LOW_SPEED_TIME, LOW_SPEED_TIME);

		static const int MAX_REDIRS = 2;
		if (MAX_REDIRS)
		{
			curl_easy_setopt(handle, CURLOPT_FOLLOWLOCATION, true);
			curl_easy_setopt(handle, CURLOPT_MAXREDIRS, MAX_REDIRS);
		}
	}

	// get header info, if success set handle to content download state
	bool CurlDownloadTask::_getHeaderInfoProc(CURL *handle)
	{
		CURLcode rc = CURLE_OK;
		do
		{
			long httpResponseCode = 0;
			rc = curl_easy_getinfo(handle, CURLINFO_RESPONSE_CODE, &httpResponseCode);
			if (CURLE_OK != rc)
			{
				break;
			}
			if (200 != httpResponseCode)
			{
				char buf[256] = {0};
				sprintf(buf
					, "When request url(%s) header info, return unexcept http response code(%ld)"
					, m_requestURL.c_str()
					, httpResponseCode);
				setErrorProc(DownloadTask::ERROR_IMPL_INTERNAL, CURLE_OK, buf);
			}

			//                curl_easy_getinfo(handle, CURLINFO_EFFECTIVE_URL, &effectiveUrl);
			//                curl_easy_getinfo(handle, CURLINFO_CONTENT_TYPE, &contentType);
			double contentLen = 0;
			rc = curl_easy_getinfo(handle, CURLINFO_CONTENT_LENGTH_DOWNLOAD, &contentLen);
			if (CURLE_OK != rc)
			{
				break;
			}

			bool acceptRanges = (std::string::npos != m_header.find("Accept-Ranges")) ? true : false;

			// get current file size
			int64_t fileSize = 0;
			if (acceptRanges && m_tempFileName.length())
			{
				fileSize = ZFileUtils::Instance()->GetFileSize(m_tempFileName.c_str());
			}

			// set header info to coTask
			ZAutoLock lock(&m_mutex);
			m_totalBytesExpected = (int64_t)contentLen;
			m_acceptRanges = acceptRanges;
			if (acceptRanges && fileSize > 0)
			{
				m_totalBytesReceived = fileSize;
			}
			m_headerAchieved = true;
		} while (0);

		if (CURLE_OK != rc)
		{
			setErrorProc(DownloadTask::ERROR_IMPL_INTERNAL, rc, curl_easy_strerror(rc));
		}
		return m_headerAchieved;
	}






	//////////////////////////////////////////////////////////////////////////////////////////////////////////
	ZRefCountedPtr<Downloader> Downloader::Create(const DownloaderHints* hints)
	{
		return new ZRefCountedObject<Downloader>(hints);
	}

	Downloader::Downloader(const DownloaderHints* hints)
	:m_stoped(false)
	{
		m_hints.countOfMaxProcessingTasks=5;
		m_hints.tempFileNameSuffix=".tmp";
		m_hints.timeoutInSeconds=60*2;
		if (hints)
		{
			m_hints=*hints;
		}
		m_updateSchedule=new ZRefCountedObject<UpdateShcedule>(this);
		Schedule(m_updateSchedule.get(),schedule_selector(UpdateShcedule::OnSchedule),0);
  }

  Downloader::~Downloader()
  {
		stop();
  }

	void Downloader::onDataTaskSuccess(const DownloadTask& task,
		std::vector<unsigned char>& data)
	{
	}

	void Downloader::onFileTaskSuccess(const DownloadTask& task)
	{
	}

	void Downloader::onTaskProgress(const DownloadTask& task,
		int64_t bytesReceived,
		int64_t totalBytesReceived,
		int64_t totalBytesExpected)
	{
	}

	void Downloader::onTaskError(const DownloadTask& task,
		int errorCode,
		int errorCodeInternal,
		const std::string& errorStr)
	{

	}

	ZRefCountedPtr<DownloadTask> Downloader::createDownloadDataTask(const std::string& srcUrl, void* udata)
	{
		CurlDownloadTaskPtr task = new ZRefCountedObject<CurlDownloadTask>(this);
		task->m_requestURL=srcUrl;
		task->m_udata=udata;
		task->init(task->m_storagePath, m_hints.tempFileNameSuffix);
		addTask(task);
		run();
		return task;
	}

	ZRefCountedPtr<DownloadTask> Downloader::createDownloadFileTask(const std::string& srcUrl, const std::string& storagePath,void* udata)
	{
		CurlDownloadTaskPtr task = new ZRefCountedObject<CurlDownloadTask>(this);
		task->m_requestURL=srcUrl;
		task->m_udata=udata;
		task->m_storagePath=storagePath;
		task->init(task->m_storagePath, m_hints.tempFileNameSuffix);
		addTask(task);
		run();
		return task;
	}

	int Downloader::OnRun()
	{
		uint32_t countOfMaxProcessingTasks = this->m_hints.countOfMaxProcessingTasks;
		// init curl content
		CURLM* curlmHandle = curl_multi_init();
		std::unordered_map<CURL*,CurlDownloadTaskPtr> coTaskMap;
		int runningHandles = 0;
		CURLMcode mcode = CURLM_OK;
		int rc = 0;                 // select return code
		ZThreadID thisThreadId = ZCurrentThreadID();
		do
		{
			// check the thread should exit or not
			{
				ZAutoLock lock(&m_threadMutex);
				// if the Impl stoped, this->_thread.reset will be called, thus _thread.get_id() not equal with thisThreadId
				if (thisThreadId != m_thread->GetId())
				{
					break;
				}
			}

			if (runningHandles)
			{
				// get timeout setting from multi-handle
				long timeoutMS = -1;
				curl_multi_timeout(curlmHandle, &timeoutMS);

				if(timeoutMS < 0)
				{
					timeoutMS = 1000;
				}

				/* get file descriptors from the transfers */
				fd_set fdread;
				fd_set fdwrite;
				fd_set fdexcep;
				int maxfd = -1;

				FD_ZERO(&fdread);
				FD_ZERO(&fdwrite);
				FD_ZERO(&fdexcep);

				mcode = curl_multi_fdset(curlmHandle, &fdread, &fdwrite, &fdexcep, &maxfd);
				if (CURLM_OK != mcode)
				{
					break;
				}

				// do wait action
				if(maxfd == -1)
				{
                    SDL_Delay(timeoutMS);
					rc = 0;
				}
				else
				{
					struct timeval timeout;

					timeout.tv_sec = timeoutMS / 1000;
					timeout.tv_usec = (timeoutMS % 1000) * 1000;

					rc = select(maxfd+1, &fdread, &fdwrite, &fdexcep, &timeout);
				}

				if (rc < 0)
				{
					LOG(LS_ERROR)<<"select return unexpect code: "<<rc;
				}
			}

			if (coTaskMap.size())
			{
				mcode = CURLM_CALL_MULTI_PERFORM;
				while(CURLM_CALL_MULTI_PERFORM == mcode)
				{
					mcode = curl_multi_perform(curlmHandle, &runningHandles);
				}
				if (CURLM_OK != mcode)
				{
					break;
				}

				struct CURLMsg *m;
				do {
					int msgq = 0;
					m = curl_multi_info_read(curlmHandle, &msgq);
					if(m && (m->msg == CURLMSG_DONE))
					{
						CURL *curlHandle = m->easy_handle;
						CURLcode errCode = m->data.result;

						CurlDownloadTaskPtr task = coTaskMap[curlHandle];

						// remove from multi-handle
						curl_multi_remove_handle(curlmHandle, curlHandle);
						bool reinited = false;
						do
						{
							if (CURLE_OK != errCode)
							{
								task->setErrorProc(DownloadTask::ERROR_IMPL_INTERNAL, errCode, curl_easy_strerror(errCode));
								break;
							}

							// if the task is content download task, cleanup the handle
							if (task->m_headerAchieved)
							{
								break;
							}

							// the task is get header task
							// first, we get info from response
							if (false ==task->_getHeaderInfoProc(curlHandle))
							{
								// the error info has been set in _getHeaderInfoProc
								break;
							}

							// after get header info success
							// wrapper.second->_totalBytesReceived inited by local file size
							// if the local file size equal with the content size from header, the file has downloaded finish
							if (task->m_totalBytesReceived &&
								task->m_totalBytesReceived == task->m_totalBytesExpected)
							{
								// the file has download complete
								// break to move this task to finish queue
								break;
							}
							// reinit curl handle for download content
							curl_easy_reset(curlHandle);
							task->_initCurlHandleProc(curlHandle, true);
							mcode = curl_multi_add_handle(curlmHandle, curlHandle);
							if (CURLM_OK != mcode)
							{
								task->setErrorProc(DownloadTask::ERROR_IMPL_INTERNAL, mcode, curl_multi_strerror(mcode));
								break;
							}
							reinited = true;
						} while (0);

						if (reinited)
						{
							continue;
						}
						curl_easy_cleanup(curlHandle);
						LOG(LS_INFO)<<" _threadProc task clean cur handle :errCode:"<<errCode;

						// remove from coTaskMap
						coTaskMap.erase(curlHandle);

						// remove from _processSet
						{
							ZAutoLock lock(&m_processMutex);
							if (m_processSet.end() != m_processSet.find(task)) {
								m_processSet.erase(task);
							}
						}

						// add to finishedQueue
						{
							ZAutoLock lock(&m_finishedMutex);
							m_finishedQueue.push_back(task);
						}
					}
				} while(m);
			}

			// process tasks in _requestList
			while (0 == countOfMaxProcessingTasks || coTaskMap.size() < countOfMaxProcessingTasks)
			{
				// get task wrapper from request queue
				CurlDownloadTaskPtr task;
				{
					ZAutoLock lock(&m_requestMutex);
					if (m_requestQueue.size())
					{
						task = static_cast<CurlDownloadTask*>(m_requestQueue.front().get());
						m_requestQueue.pop_front();
					}
				}

				// if request queue is empty, the wrapper.first is nullptr
				if (! task)
				{
					break;
				}

                if(task->GetErrorCode()==0)
				    task->initProc();

				// create curl handle from task and add into curl multi handle
				CURL* curlHandle = curl_easy_init();

				if (nullptr == curlHandle)
				{
					task->setErrorProc(DownloadTask::ERROR_IMPL_INTERNAL, 0, "Alloc curl handle failed.");
					ZAutoLock lock(&m_finishedMutex);
					m_finishedQueue.push_back(task);
					continue;
				}

				// init curl handle for get header info
				task->_initCurlHandleProc(curlHandle);

				// add curl handle to process list
				mcode = curl_multi_add_handle(curlmHandle, curlHandle);
				if (CURLM_OK != mcode)
				{
					task->setErrorProc(DownloadTask::ERROR_IMPL_INTERNAL, mcode, curl_multi_strerror(mcode));
					ZAutoLock lock(&m_finishedMutex);
					m_finishedQueue.push_back(task);
					continue;
				}

				coTaskMap[curlHandle] = task;
				ZAutoLock lock(&m_processMutex);
				m_processSet.insert(task);
			}
		} while (coTaskMap.size() && !m_stoped);

		curl_multi_cleanup(curlmHandle);
		this->stop();
		return 0;
	}

	void Downloader::addTask(DownloadTaskPtr task)
	{
		if (DownloadTask::ERROR_NO_ERROR == task->m_errCode)
		{
			ZAutoLock lock(&m_requestMutex);
			m_requestQueue.push_back(task);
		}
		else
		{
			ZAutoLock lock(&m_finishedMutex);
			m_finishedQueue.push_back(task);
		}
	}

	void Downloader::run()
	{
		ZAutoLock lock(&m_threadMutex);
		if (!m_thread)
		{
			m_stoped=false;
			m_thread=ZThread::CreateThread("Downloader");
			m_thread->Start(this);
		}
	}

	void Downloader::stop()
	{
		ZAutoLock lock(&m_threadMutex);
		m_stoped=true;
		if (m_thread)
		{
			m_thread->Join();
			m_thread=NULL;
		}
	}

	bool Downloader::stoped()
	{
		return m_stoped;
	}


	void Downloader::getProcessTasks(std::vector<DownloadTaskPtr>& outList)
	{
		ZAutoLock lock(&m_processMutex);
		outList.reserve(m_processSet.size());
		outList.insert(outList.end(), m_processSet.begin(), m_processSet.end());
	}

	void Downloader::getFinishedTasks(std::vector<DownloadTaskPtr>& outList)
	{
		ZAutoLock lock(&m_finishedMutex);
		outList.reserve(m_finishedQueue.size());
		outList.insert(outList.end(), m_finishedQueue.begin(), m_finishedQueue.end());
		m_finishedQueue.clear();
	}

	void Downloader::_onSchedule(Uint32)
	{
		std::vector<DownloadTaskPtr> tasks;
		getProcessTasks(tasks);
		for (size_t i=0;i<tasks.size();++i)
		{
			CurlDownloadTask* task = static_cast<CurlDownloadTask*>(tasks[i].get());

			ZAutoLock lock(&task->m_mutex);
			if (task->m_bytesReceived)
			{
				onTaskProgress(*task,
					task->m_bytesReceived,
					task->m_totalBytesReceived,
					task->m_totalBytesExpected);
				task->m_bytesReceived = 0;
			}
		}
		tasks.resize(0);

		// update finished tasks
		getFinishedTasks(tasks);
		if (stoped())
		{
			//ZScheduler::Unschedule(this);
		}

		for (size_t i=0;i<tasks.size();++i)
		{
			CurlDownloadTask* task = static_cast<CurlDownloadTask*>(tasks[i].get());

			// if there is bytesReceived, call progress update first
			if (task->m_bytesReceived)
			{
				onTaskProgress(*task,
					task->m_bytesReceived,
					task->m_totalBytesReceived,
					task->m_totalBytesExpected);
				task->m_bytesReceived = 0;
			}

			// if file task, close file handle and rename file if needed
			if (task->m_fp)
			{
				fclose(task->m_fp);
				task->m_fp = NULL;
				do
				{
					if (0 == task->m_fileName.length())
					{
						break;
					}

					ZFileUtils* util = ZFileUtils::Instance();
					// if file already exist, remove it
					if (util->IsFileExist(task->m_fileName.c_str()))
					{
						if (false == util->RemoveFile(task->m_fileName.c_str()))
						{
							task->m_errCode = DownloadTask::ERROR_FILE_OP_FAILED;
							task->m_errCodeInternal = 0;
							task->m_errDescription = "Can't remove old file: ";
							task->m_errDescription.append(task->m_fileName);
							break;
						}
					}

					// rename file
					if (util->RenameFile(task->m_tempFileName.c_str(), task->m_fileName.c_str()))
					{
						// success, remove storage from set
						break;
					}
					// failed
					task->m_errCode = DownloadTask::ERROR_FILE_OP_FAILED;
					task->m_errCodeInternal = 0;
					task->m_errDescription = "Can't renamefile from: ";
					task->m_errDescription.append(task->m_tempFileName);
					task->m_errDescription.append(" to: ");
					task->m_errDescription.append(task->m_fileName);
				} while (0);

			}
			// needn't lock coTask here, because tasks has removed form _impl
			if(task->m_errCode==DownloadTask::ERROR_NO_ERROR)
			{
				if(task->m_storagePath.empty())
					onDataTaskSuccess(*task, task->m_buf);
				else
					onFileTaskSuccess(*task);
			}
			else
				onTaskError(*task,task->m_errCode,task->m_errCodeInternal,task->m_errDescription);
		}
	}

}  //  namespace ZuiLib

